#include "common_task.h"
#include <iostream>
#include <algorithm>

namespace El {
namespace Common {

Task& Task::GetInstance() {
    static Task instance;
    return instance;
}

Task::Task() {
    // 初始化线程池
    initializeThreadPool(2);
}

Task::~Task() {
    {
        std::unique_lock<std::mutex> lock(queueMutex_);
        stop_ = true;
    }
    condition_.notify_all();
    for (std::thread &worker : workers_) {
        worker.join();
    }
}

std::future<void> Task::PushTask(const std::string& taskName, const TaskFunction& taskFunc) {
    auto task = std::make_shared<std::packaged_task<void()>>(taskFunc);
    std::future<void> res = task->get_future();
    {
        std::unique_lock<std::mutex> lock(queueMutex_);
        taskQueues_[taskName].push({taskName, [task]() { (*task)(); }});
    }
    condition_.notify_one();
    return res;
}

void Task::SetThreadPoolSize(size_t size) {
    initializeThreadPool(size);
}

void Task::initializeThreadPool(size_t size) {
    {
        std::unique_lock<std::mutex> lock(queueMutex_);
        stop_ = true;
    }
    condition_.notify_all();
    for (std::thread &worker : workers_) {
        worker.join();
    }
    workers_.clear();
    stop_ = false;
    for (size_t i = 0; i < size; ++i) {
        workers_.emplace_back(&Task::workerThread, this);
    }
}

void Task::workerThread() {
    while (true) {
        TaskItem task;
        std::string taskName;
        {
            std::unique_lock<std::mutex> lock(queueMutex_);
            condition_.wait(lock, [this] {
                return stop_ || std::any_of(taskQueues_.begin(), taskQueues_.end(), 
                    [this](const auto& pair) { 
                        return !pair.second.empty() && processingTasks_.find(pair.first) == processingTasks_.end();
                    });
            });
            if (stop_ && std::all_of(taskQueues_.begin(), taskQueues_.end(), 
                                     [](const auto& pair) { return pair.second.empty(); })) {
                return;
            }
            for (auto& queue : taskQueues_) {
                if (!queue.second.empty() && processingTasks_.find(queue.first) == processingTasks_.end()) {
                    taskName = queue.first;
                    task = std::move(queue.second.front());
                    queue.second.pop();
                    processingTasks_.insert(taskName);
                    break;
                }
            }
        }
        
        if (task.func) {
            // 执行当前任务
            task.func();
            
            // 任务执行完毕，processingTasks_处理中任务集合中移除
            {
                std::unique_lock<std::mutex> lock(queueMutex_);
                processingTasks_.erase(taskName);
            }
            
            // 通知其他线程，可能有新的任务可以处理
            condition_.notify_one();
        }
    }
}

} // namespace Common
} // namespace El